package com.bocommlife.mi

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.util.LongAccumulator
import org.apache.spark.util.AccumulatorV2

object SimpleAccumulator {
  private var accumulator: LongAccumulator = _
  //  private var accumulator: AccumulatorV2[Long, Long] = _
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("streaming couting testing app").set("spark.yarn.jars", "hdfs://master:9000/spark_jars/*")
    val sc = new StreamingContext(conf, Seconds(1))
    println("+++++++++++++++++++ initializing accumulator +++++++++++++++++++")
    //    accumulator = sc.sparkContext.longAccumulator("abc accum")
    accumulator = sc.sparkContext.longAccumulator("accum")
    val words = sc.socketTextStream("129.1.9.46", 9999, StorageLevel.MEMORY_AND_DISK)
    val r = words.map(word => {
      println("+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++")
      accumulator.add(1)
      println("accumulator count is..." + accumulator.value)
    })

    r.print()

    sc.start()
    sc.awaitTermination()
  }
}